package com.jzsec.tzdslog;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.time.LocalDate;
import java.time.ZoneId;

public class CCustLoginHisSource extends RichSourceFunction<CCustLoginHisEntity> {
	Logger log = LoggerFactory.getLogger(this.getClass());

	private PreparedStatement ps;
	private Connection connection;
	private String url, username, password;
	private int year;
	private Date beginDate, endDate;


	public CCustLoginHisSource(String url, String username, String password, LocalDate localDate) {
		this.url = url;
		this.username = username;
		this.password = password;

		year = localDate.getYear();
		beginDate = new java.sql.Date(
				localDate.atStartOfDay().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
		endDate = new java.sql.Date(localDate.plusDays(1).atStartOfDay().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
	}

	private Connection getConnection() {
		Connection conn = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			conn = DriverManager.getConnection(url, username, password);
		} catch (Exception e) {
			System.out.println("-----------mysql getConnection() has exception , msg = " + e.getMessage());
		}
		return conn;
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		// 增量抽取数据
		super.open(parameters);
		parameters.toMap().forEach((k, v) -> System.out.println(k + ", " + v));
		connection = getConnection();
		String sql = "select * from crm.c_cust_login_his_" + year + " where create_time between ? and ?;";
		ps = this.connection.prepareStatement(sql);
		ps.setDate(1, beginDate);
		ps.setDate(2, endDate);
	}

	@Override
	public void close() throws Exception {
		super.close();
		if (null != connection) {
			connection.close();
		}
		if (null != ps) {
			ps.close();
		}
	}

	@Override
	public void cancel() {}

	@Override
	public void run(SourceContext<CCustLoginHisEntity> sourceContext) throws Exception {
		ResultSet resultSet = ps.executeQuery();
		int i;
		for (i=0; resultSet.next(); i++) {
			CCustLoginHisEntity entity = new CCustLoginHisEntity();
			entity.setId(resultSet.getInt("id"));
			entity.setCust_id(resultSet.getString("cust_id"));
			entity.setSession_id(resultSet.getString("session_id"));
			entity.setDevice_code(resultSet.getString("device_code"));
			entity.setApp_version(resultSet.getString("app_version"));
			entity.setDevice_type(resultSet.getString("device_type"));
			entity.setEnv_type(resultSet.getString("env_type"));
			entity.setApp_type(resultSet.getString("app_type"));
			entity.setTrade_url(resultSet.getString("trade_url"));
			entity.setUser_ip(resultSet.getString("user_ip"));
			entity.setUser_agent(resultSet.getString("user_agent"));
			entity.setIp_addr(resultSet.getString("ip_addr"));
			entity.setMac(resultSet.getString("mac"));
			entity.setPhone(resultSet.getString("phone"));
			entity.setImei(resultSet.getString("imei"));
			entity.setOp_station(resultSet.getString("op_station"));
			entity.setResponse_code(resultSet.getString("response_code"));
			entity.setResponse_msg(resultSet.getString("response_msg"));
			entity.setElapsed_time(resultSet.getString("elapsed_time"));
			entity.setRequest_time(new Date(resultSet.getTimestamp("request_time").getTime()));
			entity.setCreate_time(new Date(resultSet.getTimestamp("create_time").getTime()));

			sourceContext.collect(entity);
		}
		log.info("data-num:" + i);
	}
}
